package com.shujia.flink.core;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo03Parallelism {
    public static void main(String[] args) throws Exception {
        // 1、初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 指定任务的并行度：
         * 1、env.setParallelism(2);
         * 2、提交任务时指定：通过-p参数或者是在WEB界面中指定 通常使用该种方式指定并行度
         * 3、算子后通过setParallelism(3)
         * 并行度越大，任务的吞吐量越高
         * 最终并行度可以决定Slot的数量
         * 如果没有指定并行度，则会从配置文件中加载，默认是1
         * 优先级：
         *  3 > 1 > 2
         */

//        env.setParallelism(2);


//        env.setBufferTimeout(1500);

        // 2、通过nc命令模拟一个流，使用env加载流中的数据
        DataStream<String> ds = env.socketTextStream("master", 8888);

        // 3、对每行数据进行切分 将每个单词变成一行
        // 需要指定返回值的类型
        /*
         * 可以通过lambda表达式对代码进行简化
         * 如果直接使用lambda表达式要注意指定类型
         */
        DataStream<String> wordsDS = ds.flatMap((line, collector) -> {
            String[] splits = line.split(",");
            // 遍历每一个单词
            for (String word : splits) {
                // 使用collector将结果数据发送给下游
                collector.collect(word);
            }
            // java中的lambda表达式有时无法自动推断类型，需要手动指定
        }, Types.STRING);

        // 4、将每个word变成word，1这种形式
        DataStream<Tuple2<String, Integer>> kvDS = wordsDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));

        // 5、按照每个单词进行分组
        KeyedStream<Tuple2<String, Integer>, String> grpDS = kvDS.keyBy(kv -> kv.f0);

        // 6、统计单词数量
        DataStream<Tuple2<String, Integer>> wordCntDS = grpDS.sum(1);

        wordCntDS.print();

        // 启动任务
        env.execute();


    }
}
